package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.base.model.datax.DataX;
import com.sh.data.engine.domain.dataasset.service.TableProductService;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevApiNodeTypeEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevJobService;
import com.sh.data.engine.domain.datadev.offline.service.DatadevOfflineApiService;
import com.sh.data.engine.domain.integration.api.http.model.param.ApiHttpAuthParam;
import com.sh.data.engine.domain.integration.api.http.model.param.ApiHttpRequestBodyParam;
import com.sh.data.engine.domain.integration.api.http.model.param.TestApiHttpParam;
import com.sh.data.engine.domain.integration.api.http.service.impl.ApiHttpServiceImpl;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.ExpressionParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.PreviewExpressParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.TaskConfigParam;
import com.sh.data.engine.domain.integration.api.taskconfig.service.impl.TaskConfigServiceImpl;
import com.sh.data.engine.domain.integration.api.webservice.model.param.ApiWsAuthParam;
import com.sh.data.engine.domain.integration.api.webservice.model.param.ApiWsRequestBodyParam;
import com.sh.data.engine.domain.integration.api.webservice.model.param.TestApiWsParam;
import com.sh.data.engine.domain.integration.api.webservice.service.impl.ApiWsServiceImpl;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.dataasset.entity.TableProductEntity;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropEntity;
import com.sh.data.engine.repository.dao.datadev.entity.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;

import java.io.*;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;

/**
 * api
 */
@Slf4j
public class OfflineDevApiJobRunner extends OfflineDevBaseJobRunner {

    private static final String SKIP = "in ch.qos.logback";

    private volatile Process process = null;

    private TableProductService tableProductService = SpringUtil.getBean(TableProductService.class);
    private DatadevJobService offlineDevJobService = SpringUtil.getBean(DatadevJobService.class);
    private TblPropService tblPropService = SpringUtil.getBean(TblPropService.class);
    private DataSourceService dataSourceService = SpringUtil.getBean(DataSourceService.class);

    private ApiHttpServiceImpl apiHttpService = SpringUtil.getBean(ApiHttpServiceImpl.class);
    private TaskConfigServiceImpl taskConfigService = SpringUtil.getBean(TaskConfigServiceImpl.class);
    private DatadevOfflineApiService datadevOfflineApiService =
        SpringUtil.getBean(DatadevOfflineApiService.class);

    /***
     *
     * @param jobEntity
     * @param menuNodeEntity
     * @param parentNodeIdAndLineStatusMap
     * @param childNodeIdList
     * @param allJobNodeMap
     * @param executorService
     *
     * offline task
     */
    public OfflineDevApiJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
    }

    @Override
    public void cancel() {
        process.destroy();
    }

    @Override
    public int runJob() throws Exception {

        String start = String.format("节点:[%s]开始执行", menuNodeEntity.getName());
        start = formatLogWithTime(start);
        this.writeLog(Lists.newArrayList(start));

        String finishedLog = String.format("节点:[%s]结束执行", menuNodeEntity.getName());

        DatadevOfflineApiEntity offlineDevApiTask =
            datadevOfflineApiService.getDatadevOfflineApiNodeId(menuNodeEntity.getId());
        String apiNodeType = offlineDevApiTask.getApiNodeType();
        String jobJson = "";
        DatadevOfflineApiEntity.TaskConfigRequest taskConfigRequest =
            offlineDevApiTask.getTaskConfigRequest();
        if (null == taskConfigRequest) {
            throw new BusinessException("error config");
        }
        // convert task param
        TaskConfigParam taskConfigParam =
            ConvertUtil.copyProperties(taskConfigRequest, TaskConfigParam.class);
        taskConfigParam.setFieldMappings(
            ConvertUtil.copyProperties(
                taskConfigRequest.getFieldMappings(), TaskConfigParam.FieldMapping.class));
        taskConfigParam.setPartitions(
            ConvertUtil.copyProperties(
                taskConfigRequest.getPartitions(), TaskConfigParam.PartitionField.class));
        // preview data
        OfflineDevApiPreviewEntity offlineDevApiPreviewEntity =
            offlineDevApiTask.getOfflineDevApiPreviewEntity();
        if (null == offlineDevApiPreviewEntity) {
            throw new BusinessException("error config");
        }
        PreviewExpressParam previewExpressParam =
            ConvertUtil.copyProperties(offlineDevApiPreviewEntity, PreviewExpressParam.class);
        previewExpressParam.setExpressionParamList(
            ConvertUtil.copyProperties(
                offlineDevApiPreviewEntity.getExpressionParamList(), ExpressionParam.class));

        Long targetDsId = taskConfigRequest.getTargetDsId();
        String targetTable = taskConfigRequest.getTargetTableName();
        if (Objects.equals(OfflineDevApiNodeTypeEnum.WS.getNodeDesc(), apiNodeType)) {
            OfflineDevApiWsEntity offlineDevApiWsEntity = offlineDevApiTask.getOfflineDevApiWsEntity();
            TestApiWsParam testApiWsParam =
                ConvertUtil.copyProperties(offlineDevApiWsEntity, TestApiWsParam.class);
            DataX.JobContentReader jobContentReader =
                ApiWsServiceImpl.buildJobReader(testApiWsParam, previewExpressParam, taskConfigParam);
            // auth
            testApiWsParam.setApiWsAuthParam(
                ConvertUtil.copyProperties(
                    offlineDevApiWsEntity.getApiWsAuthParam(), ApiWsAuthParam.class));
            // body
            if (null != offlineDevApiWsEntity.getApiWsRequestBodyParam()) {
                ApiWsRequestBodyParam apiHttpRequestBodyParam = new ApiWsRequestBodyParam();
                apiHttpRequestBodyParam.setContentType(
                    offlineDevApiWsEntity.getApiWsRequestBodyParam().getContentType());
                apiHttpRequestBodyParam.setBody(offlineDevApiWsEntity.getApiWsRequestBodyParam().getBody());
                if (CollectionUtils.isNotEmpty(
                    offlineDevApiWsEntity.getApiWsRequestBodyParam().getFormParams())) {
                    apiHttpRequestBodyParam.setFormParams(
                        ConvertUtil.copyProperties(
                            offlineDevApiWsEntity.getApiWsRequestBodyParam().getFormParams(),
                            ApiWsRequestBodyParam.FormParam.class));
                }
                testApiWsParam.setApiWsRequestBodyParam(apiHttpRequestBodyParam);
            }
            jobJson = taskConfigService.buildJobJson(jobContentReader, taskConfigParam);
        } else if (Objects.equals(OfflineDevApiNodeTypeEnum.HTTP.getNodeDesc(), apiNodeType)) {
            OfflineDevApiHttpEntity offlineDevApiHttpEntity =
                offlineDevApiTask.getOfflineDevApiHttpEntity();
            TestApiHttpParam testApiHttpParam =
                ConvertUtil.copyProperties(offlineDevApiHttpEntity, TestApiHttpParam.class);
            // auth
            testApiHttpParam.setApiHttpAuthParam(
                ConvertUtil.copyProperties(
                    offlineDevApiHttpEntity.getApiHttpAuthParam(), ApiHttpAuthParam.class));

            // body
            if (null != offlineDevApiHttpEntity.getApiHttpRequestBodyParam()) {
                ApiHttpRequestBodyParam apiHttpRequestBodyParam = new ApiHttpRequestBodyParam();
                apiHttpRequestBodyParam.setContentType(
                    offlineDevApiHttpEntity.getApiHttpRequestBodyParam().getContentType());
                apiHttpRequestBodyParam.setBody(
                    offlineDevApiHttpEntity.getApiHttpRequestBodyParam().getBody());
                if (CollectionUtils.isNotEmpty(
                    offlineDevApiHttpEntity.getApiHttpRequestBodyParam().getFormParams())) {
                    apiHttpRequestBodyParam.setFormParams(
                        ConvertUtil.copyProperties(
                            offlineDevApiHttpEntity.getApiHttpRequestBodyParam().getFormParams(),
                            ApiHttpRequestBodyParam.FormParam.class));
                }
                testApiHttpParam.setApiHttpRequestBodyParam(apiHttpRequestBodyParam);
            }
            DataX.JobContentReader jobContentReader =
                apiHttpService.buildJobReader(testApiHttpParam, previewExpressParam, taskConfigParam);
            jobJson = taskConfigService.buildJobJson(jobContentReader, taskConfigParam);
        } else {
            throw new BusinessException("error apiNodeType");
        }

        final String dir = fileStorageConfiguration.getDir();
        String jsonPath =
            String.join(
                File.separator,
                dataXConfig.getHome(),
                dir,
                Constants.LogFileExp.OFFLINE_DEV_JOB_LOG_EXP,
                menuNodeEntity.getId() + ".json");
        File file = new File(jsonPath);
        FileUtils.writeStringToFile(file, jobJson, "UTF-8");

        String command = dataXConfig.getCmd(jsonPath);
        log.info("cmd:{}", command);

        process = Runtime.getRuntime().exec(command);

        read(process.getInputStream());
        read(process.getErrorStream());

        int code = process.waitFor();
        log.info("exitCode:{}", code);
        this.writeLog(Lists.newArrayList(finishedLog));

        // 如果运行成功且目标数据源是系统默认的
        DataSourceDetailDomain dataSourceDetailDomain =
            dataSourceService.getDataSourceDetailById(targetDsId, false);
        if (code == 0) {
            // 产出信息记录
            if (null != dataSourceDetailDomain.getIsSystem()
                && dataSourceDetailDomain.getIsSystem().equals(1)
                && Objects.nonNull(jobEntity.getTaskFlowJobId())) {
                TblPropEntity tblPropByTblName =
                    tblPropService.getTblPropByTblName(jobEntity.getProjectId(), targetTable);
                if (tblPropByTblName != null) {
                    TableProductEntity tableProductEntity = new TableProductEntity();
                    DatadevJobEntity jobServiceById =
                        offlineDevJobService.findById(jobEntity.getTaskFlowJobId());
                    offlineDevJobService.findById(jobEntity.getTaskFlowJobId());
                    tableProductEntity.setJobId(jobEntity.getId());
                    tableProductEntity.setTaskFlowId(jobServiceById.getNodeId());
                    tableProductEntity.setTblId(tblPropByTblName.getTblId());
                    tableProductEntity.setProjectId(jobEntity.getProjectId());
                    tableProductEntity.setSubTaskId(jobEntity.getNodeId());
                    tableProductService.insetTableProduct(Lists.newArrayList(tableProductEntity));
                }
            }
            // 任务执行完成后设置任务为已完成
            return OfflineDevJobStatusEnum.SUCCESS.getCode();
        } else if (code == 143) {
            return OfflineDevJobStatusEnum.CANCEL.getCode();
        }
        return OfflineDevJobStatusEnum.FAILURE.getCode();
    }

    private void read(InputStream in) {
        new Thread(
            () -> {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, "UTF-8"))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains(SKIP)) {
                            continue;
                        }
                        List<String> logs = Lists.newArrayList(line);
                        writeLog(logs);
                    }
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
            })
            .start();
    }
}
